package mse.tx.arch.pipes;

import java.io.PipedInputStream;
import java.io.PipedOutputStream;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

import mse.FilterFramework;

public abstract class AbstractHashedMultiPortFilter extends FilterFramework {
	// Define filter input and output ports

	// [ysahn] This map holds the mapping of id to Filter-Pipes 
	private Map<String, InputFilterPipeEntry> inputFilterPorts = Collections.synchronizedMap(new HashMap<String, InputFilterPipeEntry>());
	
	//
	private Map<String, PipedOutputStream> outputPorts = Collections.synchronizedMap(new HashMap<String, PipedOutputStream>());

	// The following reference to a filter is used because java pipes are able to reliably
	// detect broken pipes on the input port of the filter. This variable will point to
	// the previous filter in the network and when it dies, we know that it has closed its
	// output pipe and will send no more data.

	public PipedOutputStream getOutputWritePort(String id)
	{
		return outputPorts.get(id);
	}

	/***************************************************************************
	* CONCRETE METHOD:: connect
	* Purpose: This method connects filters to each other. All connections are
	* through the input port of each filter. That is each filter's input port is
	* connected to another filter's output port through this method.
	* Different from the parent's single port structure, this method  
	*
	* Arguments:
	* 	@param inputPortId  the input port id to which the input filter connects to
	*   @param inFilter  the actual input filter
	*   @param inputFilterOutputPort  if the input filter is also a multi port
	*
	* Returns: void
	*
	* Exceptions: IOException
	*
	****************************************************************************/

	void connect( String inputPortId, FilterFramework inFilter, String inputFilterOutputPort )
	{
		try
		{
			// Connect this filter's named input to the upstream pipe's output stream
			InputFilterPipeEntry inputFilterPipeEntry = new InputFilterPipeEntry(inFilter);
			inputFilterPorts.put(inputPortId, inputFilterPipeEntry);

			if (inFilter instanceof AbstractHashedMultiPortFilter) {
				AbstractHashedMultiPortFilter mpInFilter = (AbstractHashedMultiPortFilter)inFilter;
				inputFilterPipeEntry.getPipedInput().connect( mpInFilter.getOutputWritePort(inputFilterOutputPort) );
			} else {
				inputFilterPipeEntry.getPipedInput().connect( inFilter.getOutputWritePort() );
			}

		} // try

		catch( Exception Error )
		{
			System.out.println( "\n" + this.getName() + " FilterFramework error connecting::"+ Error );

		} // catch

	} // Connect

	/***************************************************************************
	* CONCRETE METHOD:: ReadFilterInputPort
	* Purpose: This method reads data from the input port one byte at a time.
	*
	* Arguments: void
	*
	* Returns: byte of data read from the input port of the filter.
	*
	* Exceptions: IOExecption, EndOfStreamException (rethrown)
	*
	****************************************************************************/

	byte readFilterInputPort(String portId) throws EndOfStreamException
	{
		byte datum = 0;
		
		InputFilterPipeEntry inputPort = inputFilterPorts.get(portId);
		

		/***********************************************************************
		* Since delays are possible on upstream filters, we first wait until
		* there is data available on the input port. We check,... if no data is
		* available on the input port we wait for a quarter of a second and check
		* again. Note there is no timeout enforced here at all and if upstream
		* filters are deadlocked, then this can result in infinite waits in this
		* loop. It is necessary to check to see if we are at the end of stream
		* in the wait loop because it is possible that the upstream filter completes
		* while we are waiting. If this happens and we do not check for the end of
		* stream, then we could wait forever on an upstream pipe that is long gone.
		* Unfortunately Java pipes do not throw exceptions when the input pipe is
		* broken. So what we do here is to see if the upstream filter is alive.
		* if it is, we assume the pipe is still open and sending data. If the
		* filter is not alive, then we assume the end of stream has been reached.
		***********************************************************************/

		
		try
		{
			while (inputPort.getPipedInput().available()==0 )
			{
				if (inputPort.getInputFilter().isAlive()) // [ysahn] Check if the input filter thread is alive
				{
					throw new EndOfStreamException("End of input stream reached");

				} //if

				sleep(250);

			} // while

		} // try

		catch( EndOfStreamException Error )
		{
			throw Error;

		} // catch

		catch( Exception Error )
		{
			System.out.println( "\n" + this.getName() + " Error in read port wait loop::" + Error );

		} // catch

		/***********************************************************************
		* If at least one byte of data is available on the input
		* pipe we can read it. We read and write one byte to and from ports.
		***********************************************************************/

		try
		{
			datum = (byte)inputPort.getPipedInput().read();
			return datum;

		} // try

		catch( Exception Error )
		{
			System.out.println( "\n" + this.getName() + " Pipe read error::" + Error );
			return datum;

		} // catch

	} // ReadFilterPort

	/***************************************************************************
	* CONCRETE METHOD:: WriteFilterOutputPort
	* Purpose: This method writes data to the output port one byte at a time.
	*
	* Arguments:
	* 	byte datum - This is the byte that will be written on the output port.of
	*	the filter.
	*
	* Returns: void
	*
	* Exceptions: IOException
	*
	****************************************************************************/

	void WriteFilterOutputPort(String outPortId, byte datum)
	{
		PipedOutputStream outPort = outputPorts.get(outPortId);
		try
		{
			outPort.write((int) datum );
			outPort.flush();

		} // try

		catch( Exception Error )
		{
			System.out.println("\n" + this.getName() + " Pipe write error::" + Error );

		} // catch

		return;

	} // WriteFilterPort


	// This function is deprecated.
	//private boolean EndOfInputStream()

	/***************************************************************************
	* CONCRETE METHOD:: ClosePorts
	* Purpose: This method is used to close the input and output ports of the
	* filter. It is important that filters close their ports before the filter
	* thread exits.
	*
	* Arguments: void
	*
	* Returns: void
	*
	* Exceptions: IOExecption
	*
	****************************************************************************/

	public void ClosePorts()
	{
		// Close input ports
		Collection<InputFilterPipeEntry> filterPortEntries = inputFilterPorts.values();
		Iterator<InputFilterPipeEntry> itIfpe = filterPortEntries.iterator();
		
		while (itIfpe.hasNext()) {
			InputFilterPipeEntry obj = itIfpe.next();
			try
			{
				obj.getPipedInput().close();
			}
			catch( Exception Error )
			{
				System.out.println( "\n" + this.getName() + " ClosePorts error::" + Error );
			} // catch
		}

		// Close output ports
		Collection<PipedOutputStream> outputPortEntries = this.outputPorts.values();
		Iterator<PipedOutputStream> itPo = outputPortEntries.iterator();
		
		while (itPo.hasNext()) {
			PipedOutputStream obj = itPo.next();
			try
			{
				obj.close();
			}
			catch( Exception Error )
			{
				System.out.println( "\n" + this.getName() + " ClosePorts error::" + Error );
			} // catch
		}

	} // ClosePorts

	/***************************************************************************
	* CONCRETE METHOD:: run
	* Purpose: This is actually an abstract method defined by Thread. It is called
	* when the thread is started by calling the Thread.start() method. In this
	* case, the run() method should be overridden by the filter programmer using
	* this framework superclass
	*
	* Arguments: void
	*
	* Returns: void
	*
	* Exceptions: IOExecption
	*
	****************************************************************************/

	public void run()
    {
		// The run method should be overridden by the subordinate class. Please
		// see the example applications provided for more details.

	} // run
	
	/**
	 * Structure that contain the InputFilter - Pipe pair
	 * @author ysahn
	 *
	 */
	class InputFilterPipeEntry
	{
		
		FilterFramework inputFilter;
		PipedInputStream pipedInput = new PipedInputStream();
		
		public InputFilterPipeEntry(FilterFramework filter)
		{
			this.inputFilter = filter;
		}
		
		public FilterFramework getInputFilter() {
			return inputFilter;
		}
		public void setInputFilter(FilterFramework inputFilter) {
			this.inputFilter = inputFilter;
		}
		public PipedInputStream getPipedInput() {
			return pipedInput;
		}
		public void setPipedInput(PipedInputStream pipedInput) {
			this.pipedInput = pipedInput;
		}
		
	}

}
